package cn.iocoder.boot.springai001.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 统一的流式对话取消服务。
 * 通过 requestId（优先）或 sessionId 建立取消信号通道，
 * 在流上使用 takeUntilOther(signalFlux) 实现显式中止。
 */
@Slf4j
@Service
public class CancellationService {

    private final Map<String, Sinks.Many<Void>> keyToSink = new ConcurrentHashMap<>();

    /**
     * 获取（或创建）对应 key 的取消信号 Flux，用于在流上绑定 takeUntilOther。
     */
    public Flux<Void> signalFlux(String key) {
        if (key == null || key.isEmpty()) {
            return Flux.never();
        }
        Sinks.Many<Void> sink = keyToSink.computeIfAbsent(key, k -> Sinks.many().multicast().directBestEffort());
        return sink.asFlux();
    }

    /**
     * 发出取消信号。返回是否成功发出至少一个信号。
     */
    public boolean cancel(String key) {
        if (key == null || key.isEmpty()) {
            return false;
        }
        Sinks.Many<Void> sink = keyToSink.get(key);
        if (sink == null) {
            log.debug("取消信号未找到，key={}", key);
            return false;
        }
        Sinks.EmitResult result = sink.tryEmitComplete();
        if (result.isSuccess()) {
            log.info("已发出取消信号，key={}", key);
            return true;
        }
        log.warn("取消信号发出失败，key={}, result={}", key, result);
        return false;
    }

    /**
     * 清理对应 key 的资源。
     */
    public void cleanup(String key) {
        if (key == null || key.isEmpty()) {
            return;
        }
        keyToSink.remove(key);
    }
}


